/*
 * @Author: ygnnli 1585923072@qq.com
 * @Date: 2024-08-15 15:45:51
 * @LastEditors: ygnnli 1585923072@qq.com
 * @LastEditTime: 2024-08-27 10:47:58
 * @FilePath: /bitrpc/source/net.hpp
 * @Description: 这是默认设置,请设置`customMade`, 打开koroFileHeader查看配置 进行设置: https://github.com/OBKoro1/koro1FileHeader/wiki/%E9%85%8D%E7%BD%AE
 */
#pragma once
#include <muduo/net/TcpServer.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/TcpConnection.h>
#include <muduo/net/Buffer.h>
#include <muduo/net/TcpClient.h>
#include <muduo/base/CountDownLatch.h>
#include <muduo/net/EventLoopThread.h>
#include "abstract.hpp"
#include "detail.hpp"
#include "fields.hpp"
#include "message.hpp"
#include <mutex>
#include <unordered_map>

namespace bitrpc
{
    class MuduoBuffer : public BaseBuffer
    {
    public:
        using ptr = std::shared_ptr<MuduoBuffer>;
        MuduoBuffer(muduo::net::Buffer *buf) : _buf(buf) {}
        // 缓冲区数据的个数
        virtual size_t readableSize() override
        {
            return _buf->readableBytes();
        }

        // 尝试从缓冲区读取一个int32_t(4字节)类型的数据，没有删除！！！
        virtual int32_t peekInt32() override
        {
            // muduo库是一个网络库，从缓冲区取出一个四字节的数据，会自动转换为网络字节序的转换   把uint32_t类型从网络序转换到主机序
            return _buf->peekInt32();
        }

        // 删除缓冲区一个int32_t(4字节)类型的数据
        virtual void retriveInt32() override
        {
            return _buf->retrieveInt32();
        }

        // 从缓冲区读取一个int32_t(4字节)类型的数据并删除缓冲区的这4字节数据
        virtual int32_t readInt32() override
        {
            return _buf->readInt32();
        }

        // 从缓冲区取出指定长度的数据
        virtual std::string retrieveAsString(size_t length) override
        {
            return _buf->retrieveAsString(length);
        }

    private:
        muduo::net::Buffer *_buf;
    };

    class BufferFactory
    {
    public:
        template <typename... Args>
        static BaseBuffer::ptr create(Args &&...args)
        {
            return std::make_shared<MuduoBuffer>(std::forward<Args>(args)...); // std::forward是完美转发，保留参数类型
        }
    };

    class LVProtocol : public BaseProtocol
    {
    public:
        using ptr = std::shared_ptr<LVProtocol>;
        // 是否能够处理该消息
        // |--Len--|--VALUE--|
        // |--Len--|--mtype--|--idlen--|--id--|--body--|
        virtual bool canProcessed(const BaseBuffer::ptr &buf) override
        {
            if(buf->readableSize() < lenFieldsLength) // 如果缓冲区数据不够4字节，则不能处理
            {
                return false;
            }
            int32_t tatal_len = buf->peekInt32(); // 取出4字节的数据
            //DLOG("total_len:%d", tatal_len);
            if (buf->readableSize() < (tatal_len + lenFieldsLength)) //如果缓冲区一条数据的长度 < 消息头(len) + value ----> 消息不完整 
            {
                return false;
            }
            return true;
        }
        // 处理消息
        virtual bool onMessage(const BaseBuffer::ptr &buf, BaseMessage::ptr &msg) override
        {
            // 当调用onMessage的时候，默认认为缓冲区中的数据足够一条完整的消息
            //  |--Len(4字节)--|--mtype(4字节)--|--idlen(4字节)--|--id--|--body--|
            int32_t total_len = buf->readInt32();                                         // 读取总长度
            MType mtype = (MType)buf->readInt32();                                        // 读取消息类型
            int32_t idlen = buf->readInt32();                                             // 读取id长度
            int32_t body_len = total_len - mtypeFieldsLength - idlenFieldsLength - idlen; // 计算body长度
            std::string id = buf->retrieveAsString(idlen);                                // 读取id
            std::string body = buf->retrieveAsString(body_len);                           // 读取body 缓冲区可能有多条数据，定长获取
            msg = MessageFactory::create(mtype);                                          // 根据消息类型创建消息
            if (msg.get() == nullptr)//.get()获取指针
            {
                ELOG("消息类型错误,构造消息对象失败!");
                return false;
            }
            bool ret = msg->unserialize(body); // 反序列化
            if (ret == false)
            {
                ELOG("消息反序列化失败!");
                return false;
            }
            msg->setId(id);       // 设置id
            msg->setMType(mtype); // 设置消息类型
            return true;
        }
        // 将消息序列化
        virtual std::string serialize(const BaseMessage::ptr &msg) override
        {
            // |--Len(4字节)--|--mtype(4字节)--|--idlen(4字节)--|--id--|--body--|
            std::string body = msg->serialize();       // 序列化消息
            std::string id = msg->rid();               // 获取id 并转换成网络字节序
            auto mtype = htonl((int32_t)msg->mtype()); // 获取消息类型 并转换成网络字节序
            int32_t idlen = htonl(id.size());          // 获取id长度 并转换成网络字节序
            // int32_t total_len = htonl(mtypeFieldsLength + idlenFieldsLength + idlen + body.size()); // 计算总长度
            int32_t h_total_len = mtypeFieldsLength + idlenFieldsLength + id.size() + body.size(); // 计算总长度
            int32_t n_total_len = htonl(h_total_len);                                              // 转换成网络字节序
            //DLOG("h_total_len:%d", h_total_len);
            std::string result;
            result.reserve(h_total_len);
            result.append((char *)&n_total_len, lenFieldsLength); // 添加总长度
            result.append((char *)&mtype, mtypeFieldsLength);     // 添加消息类型
            result.append((char *)&idlen, idlenFieldsLength);     // 添加id长度
            result.append(id);                                    // 添加id
            result.append(body);                                  // 添加body
            return result;
        }

    private:
        const size_t lenFieldsLength = 4;   // 消息头长度
        const size_t mtypeFieldsLength = 4; // 消息类型长度
        const size_t idlenFieldsLength = 4; // 消息id长度
    };

    class ProtocolFactory
    {
    public:
        template <typename... Args>
        static BaseProtocol::ptr create(Args &&...args)
        {
            return std::make_shared<LVProtocol>(std::forward<Args>(args)...); // std::forward是完美转发，保留参数类型
        }
    };

    class MuduoConnection : public BaseConnection
    {
    public:
        using ptr = std::shared_ptr<MuduoConnection>;
        MuduoConnection(const muduo::net::TcpConnectionPtr &conn, const BaseProtocol::ptr &protocol)
            : _protocol(protocol), _conn(conn)
        {
        }
        // 发送消息
        virtual void send(const BaseMessage::ptr &msg) override
        {
            std::string body = _protocol->serialize(msg); // 序列化消息
            _conn->send(body);                            // 发送消息
        }
        // 关闭连接
        virtual void shutdown() override
        {
            _conn->shutdown();
        }
        // 连接是否是正常的
        virtual bool connected() override
        {
            return _conn->connected();
        }

    private:
        BaseProtocol::ptr _protocol;
        muduo::net::TcpConnectionPtr _conn; 
    };

    class ConnectionFactory
    {
    public:
        template <typename... Args>
        static BaseConnection::ptr create(Args &&...args)
        {
            return std::make_shared<MuduoConnection>(std::forward<Args>(args)...); // std::forward是完美转发，保留参数类型
        }
    };

    class MuduoServer : public BaseServer
    {
    public:
        using ptr = std::shared_ptr<MuduoServer>;
        MuduoServer(int port) : _server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port), "MuduoServer", muduo::net::TcpServer::kReusePort), // 创建服务器对象   监听地址 +端口 +服务器名称 +端口复用
                                _protocol(ProtocolFactory::create())
        {
        }

        // 启动服务器
        virtual void start()
        {
            // 设置连接事件的(建立和管理)回调函数
            _server.setConnectionCallback(std::bind(&MuduoServer::onConnection, this, std::placeholders::_1)); // 设置连接回调函数
            // 消息处理的回调函数
            _server.setMessageCallback(std::bind(&MuduoServer::onMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            // 启动服务器
            _server.start();  // 启动服务器 开始监听
            _baseloop.loop(); // 启动事件循环监听事件
        }

    private:
        // 回调函数: 处理连接事件
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            // 连接建立或断开
            if (conn->connected())
            {
                std::cout << "连接建立！ \n";
                auto muduo_conn = ConnectionFactory::create(conn, _protocol); // 创建连接对象
                {
                    std::unique_lock<std::mutex> lock(_mutex);       // 管理互斥锁
                    _conns.insert(std::make_pair(conn, muduo_conn)); // 将连接对象插入到map中
                }
                // 如果用户设置了回调函数，则调用回调函数
                if (_cb_connection)
                    _cb_connection(muduo_conn); // 调用回调函数
            }
            else
            {
                std::cout << "连接断开！ \n";
                BaseConnection::ptr muduo_conn;
                {
                    std::unique_lock<std::mutex> lock(_mutex); // 管理互斥锁
                    auto it = _conns.find(conn);
                    if (it == _conns.end())
                    {
                        return;
                    }
                    muduo_conn = it->second;
                    _conns.erase(it);
                }
                // 如果用户设置了回调函数，则调用回调函数
                if (_cb_close)
                    _cb_close(muduo_conn); // 调用回调函数
            }
        }

        // 回调函数: 处理消息事件
        void onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp)
        {
            //DLOG("连接有数据到来,开始处理了!");
            auto base_buf = BufferFactory::create(buf);
            while (1)
            {
                // 如果协议对象处理消息失败，则关闭连接
                if (_protocol->canProcessed(base_buf) == false)
                {
                    // 如果消息长度大于最大消息长度
                    if (base_buf->readableSize() > maxDataSize)
                    {
                        conn->shutdown(); // 关闭连接
                        ELOG("缓冲区中数据过大!关闭连接!");
                        return;
                    }
                    //DLOG("数据数量不足!");
                    break;
                }
                //DLOG("缓冲区中有数据可处理!");
                // 处理消息
                BaseMessage::ptr msg;
                bool ret = _protocol->onMessage(base_buf, msg);
                if (ret == false)
                {
                    conn->shutdown(); // 关闭连接
                    ELOG("缓冲区中数据错误!关闭连接!");
                    return;
                }
                //DLOG("消息反序列化成功!");
                BaseConnection::ptr base_conn;
                {
                    std::unique_lock<std::mutex> lock(_mutex); // 管理互斥锁
                    auto it = _conns.find(conn);
                    if (it == _conns.end())
                    {
                        conn->shutdown(); // 关闭连接
                        return;
                    }
                    base_conn = it->second; // 获取连接对象
                }
                //DLOG("调用回调函数进行消息处理!");
                // 如果用户设置了回调函数，则调用回调函数
                if (_cb_message)
                    _cb_message(base_conn, msg); // 调用回调函数
            }
        }

    private:
        muduo::net::EventLoop _baseloop;                                              // 事件循环对象 (主线程) 必须在TcpServer之前创建用EventLoop来构建TcpServer 反应堆+epoll
        muduo::net::TcpServer _server;                                                // 服务器对象
        BaseProtocol::ptr _protocol;                                                  // 协议对象
        std::mutex _mutex;                                                            // 互斥锁
        std::unordered_map<muduo::net::TcpConnectionPtr, BaseConnection::ptr> _conns; // 存储连接对象
        const size_t maxDataSize = (1 << 16);                                         // 64kb
    };

    class ServerFactory
    {
    public:
        template <typename... Args>
        static BaseServer::ptr create(Args &&...args)
        {
            return std::make_shared<MuduoServer>(std::forward<Args>(args)...); // std::forward是完美转发，保留参数类型
        }
    };

    class MuduoClient : public BaseClient
    {
    public:
        using ptr = std::shared_ptr<MuduoClient>;
        MuduoClient(const std::string &sip, int sport)
            : _protocol(ProtocolFactory::create()),
              _baseloop(_loopthread.startLoop()),// 创建线程对象，并启动线程
              _downlatch(1),
              _client(_baseloop, muduo::net::InetAddress(sip, sport), "MuduoClient") // 创建服务器对象   监听地址 +端口 +服务器名称
        {
        }
        // 连接服务器
        virtual void connect() override
        {
            //DLOG("设置回调函数,开始连接服务器!");
            // 设置连接事件的(建立和管理)回调函数
            _client.setConnectionCallback(std::bind(&MuduoClient::onConnection, this, std::placeholders::_1)); // 设置连接回调函数
            // 消息处理的回调函数
            _client.setMessageCallback(std::bind(&MuduoClient::onMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            _client.connect(); // 连接服务器
            _downlatch.wait(); // 阻塞等待连接成功
            //DLOG("连接服务器成功!");
        }
        // 关闭连接
        virtual void shutdown() override
        {
            _client.disconnect(); // 关闭连接
        }
        // 发送消息
        virtual bool send(const BaseMessage::ptr &msg) override
        {
            if (connected() == false)
            {
                ELOG("连接已断开!");
                return false;
            }
            _conn->send(msg); // 发送消息
            return true;
        }
        // 获取连接对象
        virtual BaseConnection::ptr connection() override
        {
            return _conn; // 获取连接对象
        }
        // 连接是否是正常的
        virtual bool connected() override
        {
            return (_conn && _conn->connected());
        }

    private:
        // 回调函数: 处理连接事件
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            // 连接建立或断开
            if (conn->connected())
            {
                std::cout << "连接建立！ \n";
                _downlatch.countDown();                             // 计数器减一 为0时唤醒
                _conn = ConnectionFactory::create(conn, _protocol); // 创建连接对象
            }
            else
            {
                std::cout << "连接断开！ \n";
                _conn.reset(); // 清空连接对象
            }
        }

        // 回调函数: 处理消息事件
        void onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp)
        {
            //DLOG("连接有数据到来，开始处理！");
            auto base_buf = BufferFactory::create(buf);
            while (1)
            {
                // 如果协议对象处理消息失败，则关闭连接
                if (_protocol->canProcessed(base_buf) == false)
                {
                    // 数据不足一条完整消息
                    if (base_buf->readableSize() > maxDataSize)
                    {
                        conn->shutdown(); // 关闭连接
                        ELOG("缓冲区中数据过大!关闭连接!");
                        return;
                    }
                    //DLOG("数据量不足！");
                    break;
                }
                //DLOG("缓冲区中数据可处理！");
                // 处理消息
                BaseMessage::ptr msg;
                bool ret = _protocol->onMessage(base_buf, msg);
                if (ret == false)
                {
                    conn->shutdown(); // 关闭连接
                    ELOG("缓冲区中数据错误!关闭连接!");
                    return;
                }
                //DLOG("缓冲区中数据解析完毕，调用回调函数进行处理！");
                // 如果用户设置了回调函数，则调用回调函数
                if (_cb_message)
                    _cb_message(_conn, msg); // 调用回调函数
            }
        }

    private:
        BaseConnection::ptr _conn;               // 保存连接对象
        muduo::CountDownLatch _downlatch;        // 计数器 防止没连接好就进行操作
        muduo::net::EventLoopThread _loopthread; // 事件循环线程,包含一个Thread对象和一个EventLoop对象 就自动开启循环监控--》_baseloop.loop()
        muduo::net::EventLoop *_baseloop;        // 事件循环对象 (主线程) 必须在TcpServer之前创建用EventLoop来构建TcpServer
        muduo::net::TcpClient _client;           // 服务器对象
        BaseProtocol::ptr _protocol;             // 协议对象
        const size_t maxDataSize = (1 << 16);    // 64kb
    };

    class ClientFactory
    {
    public:
        template <typename... Args>
        static BaseClient::ptr create(Args &&...args)
        {
            return std::make_shared<MuduoClient>(std::forward<Args>(args)...); // std::forward是完美转发，保留参数类型
        }
    };

}